package com.topn;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;

public class TReduce extends Reducer<Tkey, IntWritable, Text, IntWritable> {

    private Text rkey = new Text();
    private IntWritable rval = new IntWritable();

    @Override
    protected void reduce(Tkey key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        /*
        2020-10-8 38
        2020-10-8 36
        2020-10-18 34
        2020-10-20 32
         */
        Iterator<IntWritable> iter = values.iterator();

        // 是否是本组数据迭代的第一条
        boolean isFirst = true;

        // 记录第一条的day
        int day = 0;

        // 因为一组的数据day可能不同，在遍历value的时候，key的值是会变化的
        while (iter.hasNext()) {
            IntWritable temp = iter.next();

            if (isFirst) {
                rkey.set(key.getYear() + "-" + key.getMonth() + "-" + key.getDay() + "-" + key.getLocation());
                rval.set(key.getTemp());

                isFirst = false;
                day = key.getDay();
                context.write(rkey, rval);
            } else if (key.getDay() != day) {
                // 遍历的不是第一条, 并且 day和第一条也不一致
                rkey.set(key.getYear() + "-" + key.getMonth() + "-" + key.getDay() + "-" + key.getLocation());
                rval.set(key.getTemp());
                context.write(rkey, rval);
                continue;
            }
        }

    }
}
